package work.mediway.mdm.biz.helper;

import com.mediway.hos.common.core.redis.RedisCache;

import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import work.mediway.mdm.core.constant.MdmConstant;
import work.mediway.mdm.core.exception.BizException;
import work.mediway.mdm.biz.interfaces.IDataQualityPullData;

import static work.mediway.mdm.core.constant.MdmConstant.DataQuality.DEFAULT_STEP;
import static work.mediway.mdm.core.enums.DatabaseTypeEnum.CACHE;
import static work.mediway.mdm.core.enums.DatabaseTypeEnum.HIVE;
import static work.mediway.mdm.core.enums.DatabaseTypeEnum.MYSQL;
import static work.mediway.mdm.core.enums.DatabaseTypeEnum.ORACLE;
import static work.mediway.mdm.core.enums.DatabaseTypeEnum.PGSQL;

/**
 * @author zhongyj <1126834403@qq.com><br/>
 * @date 2022/7/20
 */
@SuppressWarnings({"rawtypes", "unchecked"})
@Slf4j
@Component
@RequiredArgsConstructor
public class DataPullAndCacheHelper {

    private final RedisCache redisCache;

    /**
     * 流式获取数据并缓存, 缓存格式为map
     *
     * @param connection Connection
     * @param sql        String
     * @param cacheKey   String
     * @param field      List<String>
     * @return long
     */
    public long pullDataFromDatabase(Connection connection, String sql, String cacheKey, List<String> field) {
        long totalCount = 0;
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        TimeInterval timer = DateUtil.timer();
        try {
            //根据链接创建statement 进行 sql提交   做一些statement 配置
            statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            fillPreparedStatement(connection, statement);
            //====================statement执行sql======================
            resultSet = statement.executeQuery();
            //================================================

            log.info("===================开始拉取数据===================");
            log.info("拉取数据sql ： {} ", sql);
            if (field.size() == 1) {
                List<String> tmp = CollUtil.newArrayList();
                while (resultSet.next()) {
                    tmp.add(resultSet.getString(field.get(0)));
                    if (tmp.size() == DEFAULT_STEP) {
                        redisCache.setCacheList(cacheKey, tmp);
                        log.debug(StrUtil.format("流式读取数据 sql = {}, 放入Redis缓存{}", sql, cacheKey));
                        tmp.clear();
                    }
                    totalCount++;
                }
                if (CollUtil.isNotEmpty(tmp)) {
                    redisCache.setCacheList(cacheKey, tmp);
                    tmp.clear();
                }
            } else {
                List<Map<String, Object>> tmp = CollUtil.newArrayList();
                while (resultSet.next()) {
                    Map<String, Object> data = MapUtil.newHashMap();
                    for (String item : field) {
                        data.put(item, resultSet.getString(item));
                    }
                    tmp.add(data);
                    if (tmp.size() == DEFAULT_STEP) {
                        redisCache.setCacheList(cacheKey, tmp);
                        log.debug(StrUtil.format("流式读取数据 sql = {}, 放入Redis缓存{}, 当前下标【{}】", sql, cacheKey, totalCount));
                        tmp.clear();
                    }
                    totalCount++;
                }
                if (CollUtil.isNotEmpty(tmp)) {
                    redisCache.setCacheList(cacheKey, tmp);
                    tmp.clear();
                }
            }
        } catch (SQLException e) {
            throw new BizException("获取系统数据库数据异常", e);
        } finally {
            try {
                if (resultSet != null) {
                    resultSet.close();
                }
                if (statement != null) {
                    statement.close();
                }
                connection.close();
            } catch (SQLException e) {
                log.error("获取系统数据库数据异常: {}", e.getMessage());
            }
        }
        log.info("流式读取数据库资源结束, 共读取数据{}条, 共计用时: {} ", totalCount, timer.intervalPretty());
        //先开后关原则
        return totalCount;
    }

    /**
     * 拉取数据库数据
     *
     * @param connection          Connection
     * @param sql                 String
     * @param field               取数字段
     * @param dataQualityPullData IDataQualityPullData
     */
    public void pullDataFromDatabaseAndDeal(Connection connection, String sql, List<String> field, IDataQualityPullData dataQualityPullData) {
        long totalCount = 0;
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        TimeInterval timer = DateUtil.timer();
        try {
            //根据链接创建statement 进行 sql提交   做一些statement 配置
            statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            fillPreparedStatement(connection, statement);
            //====================statement执行sql======================
            resultSet = statement.executeQuery();
            //================================================

            log.info("===================开始拉取数据===================");
            log.info("拉取数据sql ： {} ", sql);
            List<Map<String, Object>> tmp = CollUtil.newArrayList();
            while (resultSet.next()) {
                for (String item : field) {
                    Map<String, Object> data = MapUtil.newHashMap();
                    data.put(item, resultSet.getString(item));
                    tmp.add(data);
                }
                if (tmp.size() == DEFAULT_STEP) {
                    dataQualityPullData.exec(tmp);
                    tmp.clear();
                }
                totalCount++;
            }
            if (CollUtil.isNotEmpty(tmp)) {
                dataQualityPullData.exec(tmp);
                tmp.clear();
            }
        } catch (SQLException e) {
            throw new BizException("获取系统数据库数据异常", e);
        } finally {
            try {
                if (resultSet != null) {
                    resultSet.close();
                }
                if (statement != null) {
                    statement.close();
                }
                connection.close();
            } catch (Exception e) {
                e.printStackTrace();
                log.error("拉取数据库数据处理异常: {}", e.getMessage());
            }
        }
        log.info("流式读取数据库资源结束, 共读取数据{}条, 共计用时: {} ", totalCount, timer.intervalPretty());
    }

    /**
     * 从缓存中获取数据
     *
     * @param key         String
     * @param total       int
     * @param dataQuality IDataQuality
     */
    public void pullDataFormCache(String key, long total, IDataQualityPullData dataQuality) {
        for (int i = 0; i < total; i += DEFAULT_STEP) {
            log.debug(StrUtil.format("从缓存中获取系统数据, key : {}, 总数为: {}, 当前取数下标: {}", key, total, i));
            List range = redisCache.redisTemplate.opsForList().range(key, i, i + DEFAULT_STEP);
            if (CollUtil.isNotEmpty(range)) {
                dataQuality.exec(range);
            } else {
                log.info("缓存中未取到key为：{} 的数据", key);
            }
        }
    }

    /**
     * 完善PreparedStatement
     *
     * @param connection Connection
     * @param statement  PreparedStatement
     * @throws SQLException SQLException
     */
    private void fillPreparedStatement(Connection connection, PreparedStatement statement) throws SQLException {
        DatabaseMetaData md = connection.getMetaData();
        if (StrUtil.equalsAnyIgnoreCase(md.getDatabaseProductName(), MYSQL.getType())) {
            statement.setFetchSize(Integer.MIN_VALUE);
        } else if (StrUtil.equalsAnyIgnoreCase(md.getDatabaseProductName(), PGSQL.getType())) {
            connection.setAutoCommit(false);
            statement.setFetchSize(DEFAULT_STEP);
        } else if (StrUtil.equalsAnyIgnoreCase(md.getDatabaseProductName(), CACHE.getType())) {
            statement.setFetchSize(1);
        } else if (StrUtil.equalsAnyIgnoreCase(md.getDatabaseProductName(), ORACLE.getType())) {
            statement.setFetchSize(0);
        } else if (StrUtil.equalsAnyIgnoreCase(md.getDatabaseProductName(), HIVE.getType())) {
            statement.setFetchSize(DEFAULT_STEP);
        } else {
            statement.setFetchSize(Integer.MIN_VALUE);
        }
    }

    /**
     * 统计数据库数据总数
     *
     * @param connection Connection
     * @param sql        String
     * @return 总数
     */
    public long getDatabaseDataCount(Connection connection, String sql) {
        long totalCount = 0;
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        try {
            //根据链接创建statement 进行 sql提交   做一些statement 配置
            statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
            fillPreparedStatement(connection, statement);
            //====================statement执行sql======================
            resultSet = statement.executeQuery();
            //================================================
            while (resultSet.next()) {
                totalCount++;
                if (totalCount % MdmConstant.DataQuality.DEFAULT_STEP == 0) {
                    log.info("表数据量: {}", totalCount);
                }
            }
        } catch (SQLException e) {
            throw new BizException("统计数据库数量总数异常", e);
        } finally {
            try {
                if (resultSet != null) {
                    resultSet.close();
                }
                if (statement != null) {
                    statement.close();
                }
                connection.close();
            } catch (SQLException e) {
                log.error("统计数据库数量总数异常: {}", e.getMessage());
            }
        }
        //先开后关原则
        return totalCount;
    }
}
